package org.duang.db.mongo;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.bson.types.ObjectId;
import org.duang.db.IDao;
import org.duang.db.IdEntity;
import org.duang.db.common.Page;
import org.duang.db.common.Query;
import org.duang.db.common.Update;
import org.duang.db.mongo.parser.DecodeConvetor;
import org.duang.db.mongo.parser.EncodeConvetor;
import org.duang.kit.ClassKit;
import org.duang.kit.MongoKit;
import org.duang.kit.ToolsKit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.mongodb.AggregationOutput;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MapReduceCommand;
import com.mongodb.MapReduceCommand.OutputType;
import com.mongodb.MapReduceOutput;
import com.mongodb.WriteResult;

public abstract class MongoBaseDao<T> implements IDao<T> {

	private final static Logger logger = LoggerFactory.getLogger(MongoBaseDao.class);
	
	protected Class<T> cls;
	protected DB db;
	protected DBCollection coll;
	protected DBObject keys;
	
	public MongoBaseDao(){
		
	}
	
	public MongoBaseDao(Class<T> cls){
		this(null, cls);
	}
	
	public MongoBaseDao(DB database, Class<T> cls){
		boolean isExtends = ClassKit.isExtends(cls, IdEntity.class.getCanonicalName());
		if(!isExtends){
			throw new RuntimeException("the "+cls.getCanonicalName()+" is not extends "+ IdEntity.class.getCanonicalName() +", exit...");
		}
		this.cls = cls;
		db = (null == database) ? Mongo.getInstance().getDB() : database;
 		coll = db.getCollection(ClassKit.getBeanName(cls));
		keys = MongoKit.convert2DBFields(ClassKit.getFields(cls));
		try{
			MongoKit.createIndex(coll, cls);
		} catch(Exception e){
			logger.error(coll.getFullName()+" Create Index Fail: " + e.getMessage());
		}
	}
	
	/**
	 * 持久化到数据库, 会自己根据entity是否有id值进行保存或更新操作
	 * @param entity	需要持久化的对象
	 * @return			正确持久化到数据库返回true, 否则执行出异常
	 */
	public boolean save(T entity) throws Exception {
		IdEntity idEntity = (IdEntity)entity;
		if(ToolsKit.isEmpty(idEntity.getId())){
			idEntity.setId(null);
		}
		WriteResult wr = doSaveOrUpdate(idEntity);
		if (wr.getN() >= 0){
			return true;
		} else { 
			throw new RuntimeException(wr.toString());
		}
	}
	
	private WriteResult doSaveOrUpdate(IdEntity entity) throws Exception {
		DBObject dbo = EncodeConvetor.convetor(entity);
		WriteResult wr = null;
		String id = entity.getId();
		if(ToolsKit.isEmpty(id)){
			wr = coll.save(dbo);			
			entity.setId(dbo.get(IdEntity.ID_FIELD).toString());
		}
//			ToolsKit.setObjectId2Thread(entity.getId());
//		}else if(ToolsKit.isNotEmpty(id) && ToolsKit.isNotEmpty(ToolsKit.getObjectIdByThread())){
//			id = ToolsKit.getObjectIdByThread();
//			dbo.put(IdEntity.ID_FIELD, ToolsKit.message2ObjectId(id));
//			wr = coll.insert(dbo);
//			ToolsKit.removeObjectIdByThread();
//			entity.setId(dbo.get(IdEntity.ID_FIELD).toString());
//		} 
		else {
			dbo.removeField(IdEntity.ID_FIELD);
			DBObject updateDbo = new BasicDBObject(Operator.SET, dbo); 
			wr = update(id, updateDbo);
		}
		return wr;
	}
	
	public boolean insert(T entity) throws Exception {
		DBObject dbo = EncodeConvetor.convetor((IdEntity)entity);
		WriteResult wr = coll.insert(dbo);
		if (wr.getN() >= 0){
			return true;
		} else { 
			throw new RuntimeException(wr.toString());
		}
	}
	
	private WriteResult update(String id, DBObject dbo) throws Exception {
		if(!ToolsKit.isValidObjectId(id)){
			throw new IllegalArgumentException("id is not ObjectId!");
		}
		DBObject query = new BasicDBObject(IdEntity.ID_FIELD, new ObjectId(id));
		return coll.update(query, dbo, false, false);
	}
	
	public WriteResult update(Query query, Update update) throws Exception {
		/**
		 * 第三个参数代表不存在时，是否新增。
		 * 第四个参数代表是否支持批量操作。
		 */
		return coll.update(query.getQueryObj(), update.getUpdateObj(), false, true);	
	}
	
	public T findOne(Query query) throws Exception {		
		DBObject queryDbo = query.getQueryObj();		
		DBObject resultDbo = coll.findOne(queryDbo, query.getDBFields(), query.getDBOrder());		
		return DecodeConvetor.convetor(cls, resultDbo);
	}
	
	@SuppressWarnings("unchecked")
	public List<T> findList(Query query) throws Exception {
		DBObject queryObj = query.getQueryObj();
		DBObject fields =  query.getDBFields();
		DBObject sorts = query.getDBOrder();
		Page<T> page = query.getPage();
		int pageNo = page.getPageNo();
		int pageSize = page.getPageSize();
		DBCursor cursor = null;
		cursor = (ToolsKit.isNotEmpty(fields)) ?  coll.find(queryObj, fields) : coll.find(queryObj, keys);
		if(ToolsKit.isNotEmpty(sorts)) cursor.sort(sorts);
		if(pageNo>0 && pageSize>1){ cursor.skip((pageNo-1)*pageSize).limit(pageSize);}
		logger.debug("findList: " + cursor.toString());
		return dBCursor2List(cls, cursor);
	}
	
	@SuppressWarnings("unchecked")
	public Page<T> findPage(Query query) throws Exception {
		DBObject queryObj = query.getQueryObj();
		DBObject fields =  query.getDBFields();
		DBObject sorts = query.getDBOrder();
		Page<T> page = query.getPage();
		int pageNo = page.getPageNo();
		int pageSize = page.getPageSize();
		DBCursor cursor = null;
		cursor = (ToolsKit.isNotEmpty(fields)) ?  coll.find(queryObj, fields) : coll.find(queryObj, keys);
		if(ToolsKit.isNotEmpty(sorts)) cursor.sort(sorts);
		if(pageNo>0 && pageSize>0){ cursor.skip((pageNo-1)*pageSize).limit(pageSize);}
		logger.debug("findList: " + cursor.toString());
		page.setResult(dBCursor2List(cls, cursor));
		if(page.isAutoCount()){
			page.setTotalCount(count(query));
		}
		return page;
	}
	
	protected <T> List<T> dBCursor2List(Class<T> cls, DBCursor cursor){
		List<T> dboList = new ArrayList<T>(cursor.size());
		while(cursor.hasNext()){
			DBObject dbo = cursor.next();
			dboList.add(DecodeConvetor.convetor(cls, dbo));
		}
		return dboList;
	}
	
	/**
	 * 根据查询条件对多个数字字段进行求和
	 * @param keys		需要求和的字段集合
	 * @param query		查询条件
	 * @return
	 */
	public Map<String, Double> sum_mapreduct(List<String> keys, Query query) {
		if(ToolsKit.isEmpty(keys) || ToolsKit.isEmpty(query.getQueryObj())) return null;
		StringBuilder map = new StringBuilder();
		StringBuilder emit = new StringBuilder();
		StringBuilder reducedItem = new StringBuilder();
		StringBuilder reducedItem2 = new StringBuilder();
		map.append("function(){emit('");
		for(String key : keys){
           map.append(key).append(",");
           emit.append("'").append(key).append("':this.").append(key).append(",");
           reducedItem.append(key).append(":").append("0").append(",");
           reducedItem2.append("reduced.").append(key).append("+=values[i].").append(key).append(";");
		}
		map.deleteCharAt(map.length()-1);
		emit.deleteCharAt(emit.length()-1);
		reducedItem.deleteCharAt(reducedItem.length()-1);
        map.append("', {").append(emit).append("});}");
        
        StringBuilder reduce = new StringBuilder("function(key, values){");
        reduce.append("var reduced = {").append(reducedItem).append("};");
        reduce.append("for(var i=0;i<values.length; i++){");
        reduce.append(reducedItem2);
        reduce.append("} return reduced;}");
        
        logger.error("map.toString(): " + map.toString());
    	logger.error("reduce.toString(): " + reduce.toString());
    	logger.error("query.toString(): " + query.getQueryObj().toString());      
    	logger.error("sum result is null or Empty! return a Empty map!");
        
        Iterable<DBObject> results = null;
        try{
            results = mapReduce(map.toString(), reduce.toString(), query.getQueryObj());
        }catch(Exception ex){
        	ex.printStackTrace();
            logger.error(ex.getMessage(), ex);
        }
        Map<String,Double> resultMap = new HashMap<String,Double>();        
        DBObject result = null;
        if(ToolsKit.isNotEmpty(results)){
	        for(Iterator<DBObject> it = results.iterator(); it.hasNext(); ){
	        	result = it.next();	
	        }
        }
        if(null == result){        	
        	return null;
        }
        DBObject dbo = (DBObject)result.get("value");
        if(ToolsKit.isNotEmpty(dbo)){
	        for(String key : keys){
	        	resultMap.put(key, Double.parseDouble(ToolsKit.isEmpty(dbo.get(key)) ? "0" : dbo.get(key).toString().trim()));
	        }
        }
        return resultMap;
	}
	
	@SuppressWarnings("static-access")
	public Map<String, Double> sum(List<String> keys, Query query) {
		List<DBObject> pipeline = new ArrayList<DBObject>();
		DBObject matchDbo = new BasicDBObject(Operator.MATCH, query.getQueryObj());		//查询条件
		pipeline.add(matchDbo);
		DBObject maxTmp = new BasicDBObject();
		maxTmp.put("_id",null);			//返回字段
		for(String key : keys){
			DBObject sum = new BasicDBObject();
			sum.put(Operator.SUM,"$"+key);
			maxTmp.put("_"+key,sum);			//表达式
		}
		DBObject groupDbo = new BasicDBObject(Operator.GROUP, maxTmp);
		pipeline.add(groupDbo);
		Map<String,Double> resultMap = new HashMap<String,Double>();   
        try{
        	Iterable<DBObject> results = coll.aggregate(pipeline, coll.getReadPreference().secondaryPreferred()).results();
        	DBObject result = null;
            if(ToolsKit.isNotEmpty(results)){
            	result = results.iterator().next();
            }
            if(null == result){ return null; }
	        for(String key : keys){
	        	String keyTmp = "_"+key;
	        	resultMap.put(key, Double.parseDouble(ToolsKit.isEmpty(result.get(keyTmp)) ? "0" : result.get(keyTmp).toString().trim()));
	        }
        }catch(Exception ex){
            logger.error(ex.getMessage(), ex);
        }
		return resultMap;
	}
	
	/**
	 * 根据查询条件对数字字段进行求和(不支持嵌套字段查询)
	 * @param key		需要求和的字段
	 * @param query		查询条件
	 * @return
	 */
	public double sum(String key, Query query) {
		if(ToolsKit.isEmpty(key) || ToolsKit.isEmpty(query.getQueryObj())) return 0d;
		List<String> keys = new ArrayList<String>(1);
		keys.add(key);
		Map<String,Double> map = sum(keys, query);
        return (ToolsKit.isEmpty(map) || ToolsKit.isEmpty(map.get(key))) ? 0 : map.get(key);
	}
	
	 @SuppressWarnings("static-access")
	private Iterable<DBObject> mapReduce(String map, String reduce, DBObject query) throws Exception {
		MapReduceCommand command = new MapReduceCommand(coll, map, reduce, null, OutputType.INLINE, query);
		if(ToolsKit.isNotEmpty(coll.getReadPreference())) command.setReadPreference(coll.getReadPreference().secondaryPreferred());
		command.setOutputDB("{replace:'sys_mapreduce', nonAtomic:true}");
        MapReduceOutput output = coll.mapReduce(command);
        return output.results();
    }
	 
	 
	/**
	 * 根据查询条件进行汇总
	 * @param query		查询条件
	 * @return
	 */
	public long count(Query query){
		return coll.count(query.getQueryObj(), coll.getReadPreference());
	}
	
	private double max_mapeduce(String key, Query query) {
		StringBuilder map = new StringBuilder("function(){emit('");
        map.append(key);
        map.append("', {'value':this.");
        map.append(key);
        map.append("});}");
        String reduce = "function(key, values){var max=values[0].value; for(var i=1;i<values.length; i++){if(values[i].value>max){max=values[i].value;}} return {'value':max}}";
        Iterable<DBObject> results = null;
        try{
            results = mapReduce(map.toString(), reduce, query.getQueryObj());
        }catch(Exception ex){
            logger.error(ex.getMessage(), ex);
        }
        DBObject result = results.iterator().next();
        DBObject dbo = (DBObject)result.get("value");
        return Double.parseDouble(dbo.get("value").toString());
	}
	
	/**
	 * 求最大值
	 * @param key			求最大值的字段
	 * @param query			查询条件
	 * @return
	 */
	@SuppressWarnings("static-access")
	public double max(String key, Query query) {
		List<DBObject> pipeline = new ArrayList<DBObject>();
		DBObject matchDbo = new BasicDBObject(Operator.MATCH, query.getDBOrder());		//查询条件
		DBObject maxTmp = new BasicDBObject();
		maxTmp.put("_id", null);
		DBObject max = new BasicDBObject();
		max.put(Operator.MAX, "$"+key);
		maxTmp.put("_max",max);
		DBObject groupDbo = new BasicDBObject(Operator.GROUP, maxTmp);
		pipeline.add(matchDbo);
		pipeline.add(groupDbo);
		AggregationOutput out = coll.aggregate(pipeline, coll.getReadPreference().secondaryPreferred());
		try{
		    DBObject result = out.results().iterator().next();
			return Double.parseDouble(result.get("_max").toString());
		}catch(Exception ex){
		    logger.error(ex.getMessage(), ex);
		    return 0d;
		}
	}
	
	/**
	 * 求最小值
	 * @param key			求最小值的字段
	 * @param query			查询条件
	 * @return
	 */
	private double min_mapreduce(String key, Query query) {
		StringBuilder map = new StringBuilder("function(){emit('");
        map.append(key);
        map.append("', {'value':this.");
        map.append(key);
        map.append("});}");
        String reduce = "function(key, values){var min=values[0].value; for(var i=1;i<values.length; i++){if(values[i].value<min){min=values[i].value;}} return {'value':min}}";
        Iterable<DBObject> results = null;
        try{
            results = mapReduce(map.toString(), reduce, query.getQueryObj());
        }catch(Exception ex){
            logger.error(ex.getMessage(), ex);
        }
        DBObject result = results.iterator().next();
        DBObject dbo = (DBObject)result.get("value");
        return Double.parseDouble(dbo.get("value").toString());
	}
	
	/**
	 * 求最大值
	 * @param key			求最大值的字段
	 * @param query			查询条件
	 * @return
	 */
	@SuppressWarnings("static-access")
	public double min(String key, Query query) {
		List<DBObject> pipeline = new ArrayList<DBObject>();
		DBObject matchDbo = new BasicDBObject(Operator.MATCH, query.getDBOrder());		//查询条件
		DBObject minTmp = new BasicDBObject();
		minTmp.put("_id", null);
		DBObject min = new BasicDBObject();
		min.put(Operator.MIN, "$"+key);
		minTmp.put("_min",min);
		DBObject groupDbo = new BasicDBObject(Operator.GROUP, minTmp);
		pipeline.add(matchDbo);
		pipeline.add(groupDbo);
		AggregationOutput out = coll.aggregate(pipeline, coll.getReadPreference().secondaryPreferred());
		try{
		    DBObject result = out.results().iterator().next();
			return Double.parseDouble(result.get("_min").toString());
		}catch(Exception ex){
		    logger.error(ex.getMessage(), ex);
		    return 0d;
		}
	}
	
	/**
	 * 分组查询(默认按降序排序)
	 * @param key		要分组查询的字段
	 * @param query		查询条件
	 * @return
	 */
	public List<Map>  group(String key, Query query){
		return group(key, query, "desc");
	}
	/**
	 * 分组查询
	 * @param key		要分组查询的字段
	 * @param query		查询条件
	 * @param sort		结果集排序方向
	 * @return
	 */
	public List<Map>  group(String key, Query query, final String sort){
		List<String> keys = new ArrayList<String>();
		keys.add(key);
		return group(keys, query, sort);
	}
	
	/**
	 * 分组查询
	 * @param keys		要分组查询的字段集合
	 * @param query		查询条件
	 * @param sort		结果集排序方向
	 * @return
	 */
	@SuppressWarnings({ "unchecked", "static-access" })
	public List<Map> group(List<String> keys, Query query, final String sort){
		DBObject groupFields = new BasicDBObject();
		for(String key : keys){groupFields.put(key, true);}
		String reduce = "function(doc, aggr){aggr.count += 1;}";	
		DBObject dbo = null;
		try{
			dbo = coll.group(groupFields, query.getQueryObj(), new BasicDBObject("count", 0), reduce, "", coll.getReadPreference().secondaryPreferred());
		}catch (Exception ex){
			logger.error(ex.getMessage(), ex);
		}
		if(null == dbo) return null;
		java.util.List<Map> list = new ArrayList<Map>();
		for(Iterator<String> it = dbo.keySet().iterator(); it.hasNext();){
			String key = it.next();
			DBObject dboTmp = (DBObject)dbo.get(key);
			list.add(dboTmp.toMap());
		}		
		java.util.Collections.sort(list, new Comparator(){
			@Override
			public int compare(Object dbo1, Object dbo2) {
				double count1 = (Double)((Map)dbo1).get("count");
				double count2 = (Double)((Map)dbo2).get("count");
				if("desc".equals(sort))
					return (count1 > count2) ? 0 : 1;
				else
					return (count1 > count2) ? 1 : 0;
			}
		});
		return list;
	}
	
	/**
	 *集合是否存在
	 * @return	存在返回true
	 */
	public boolean isExist(){
		return coll.getDB().collectionExists(coll.getName());
	}
	
	/**
	 * 去重查询
	 * @param key		去重关键字
	 * @param query		查询条件
	 * @return			去重关键字的集合
	 */
	@SuppressWarnings({ "unchecked", "static-access" })
	public List<String> distinct(String key, Query query) {		
		return coll.distinct(key, query.getQueryObj(), coll.getReadPreference().secondaryPreferred());
	}
	
	/**
	 * 根据查询条件更新
	 * @param query   查询对象
	 * @param update  更新对象
	 */
	public void set(Query query, Update update) {
		 coll.updateMulti(query.getQueryObj(), update.getUpdateObj());
	}
	
	/**
	 * 向array/list/set添加值
	 * @param query				查询对象
	 * @param update			添加/更新对象
	 */
	@Deprecated
	public void push(Query query, Update update) {
		 coll.updateMulti(query.getQueryObj(), update.getUpdateObj());
	}
	
	/**
	 * 向array/list/set删除值
	 * @param query				查询对象
	 * @param update			删除对象
	 */
	@Deprecated
	public void pull(Query query, Update update) {
		coll.updateMulti(query.getQueryObj(), update.getUpdateObj());
	}
	
	/**
	 * 根据查询条件及分组字段统计大小
	 * @param key				要分组的字段
	 * @param query			查询条件
	 * @return
	 */
	@SuppressWarnings("static-access")
	public int groupBySize(String key, Query query){
		DBObject groupFields = new BasicDBObject();
		groupFields.put(key, true);
		String reduce = "function(doc, aggr){aggr.count += 1;}";	
		DBObject dbo = null;
		try{
			dbo = coll.group(groupFields, query.getQueryObj(), new BasicDBObject("count", 0), reduce, "", coll.getReadPreference().secondaryPreferred());
		}catch (Exception ex){
			logger.error(ex.getMessage(), ex);
		}
		if(null == dbo || null == dbo.keySet()) return 0;
		return dbo.keySet().size();
	}
	
}
